package com.jy.oms.framework.asyn.task.executor;

import com.jy.oms.framework.asyn.task.cluster.model.WorkModeEnum;
import com.jy.oms.framework.asyn.task.cluster.model.WorkNode;
import com.jy.oms.framework.asyn.task.cluster.model.WorkNodeExecutableTask;
import com.jy.oms.framework.asyn.task.cluster.registry.IRegistryService;
import com.jy.oms.framework.asyn.task.constants.CacheConstants;
import com.jy.oms.framework.asyn.task.constants.TaskQueueTypeEnum;
import com.jy.oms.framework.asyn.task.core.ExecutorContext;
import com.jy.oms.framework.asyn.task.core.RedissonClientProxy;
import com.jy.oms.framework.asyn.task.executor.bo.TaskPollResult;
import com.jy.oms.framework.asyn.task.model.TaskGroup;
import com.jy.oms.framework.asyn.task.model.TaskQueryDto;
import com.jy.oms.framework.asyn.task.model.TaskWrapper;
import org.apache.commons.collections4.CollectionUtils;
import org.apache.commons.collections4.MapUtils;
import org.apache.commons.lang3.tuple.Pair;
import org.redisson.api.RMap;
import org.redisson.api.RedissonClient;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Service;

import java.util.Comparator;
import java.util.Date;
import java.util.Map;
import java.util.concurrent.TimeUnit;

/**
 * 集群模式任务抽取器
 *
 * @author liuquan
 */
@Service
public class GroupClusterTaskExtractor extends GroupTaskExtractor {

    @Autowired
    private IRegistryService registryService;

    @Override
    public boolean ableExtract(TaskGroup taskGroup) {
        return ExecutorContext.getExecutorConfig().isEnableCluster();
    }

    @Override
    public TaskPollResult extract(TaskGroup taskGroup) {
        WorkNode workNode = registryService.getWorkNode();
        if (workNode.getWorkModel() == WorkModeEnum.SHARE) {
            Pair<Boolean, WorkNodeExecutableTask> ableCollect = workNode.ableCollectShareTask(taskGroup);
            //任务组独享模式，其他节点不能执行此任务组
            if (!ableCollect.getLeft() && CollectionUtils.isEmpty(ableCollect.getRight().getTaskCodes())) {
                return null;
            }

            if (TaskQueueTypeEnum.DB == taskGroup.getTaskQueueType()) {
                String lockKey = CacheConstants.genCacheKey(CacheConstants.TASK_EXTRACTOR_QUERY_TIME_MAP + ":lock");
                return RedissonClientProxy.lock(lockKey, () -> this.collectShareTask(taskGroup, workNode, ableCollect.getRight()));
            } else {
                return super.extract(taskGroup);
            }
        } else {
            return this.collectExclusiveTask(taskGroup, workNode);
        }
    }

    private TaskPollResult collectShareTask(TaskGroup taskGroup, WorkNode workNode, WorkNodeExecutableTask exclusiveExecutableTask) {
        String cacheKey = CacheConstants.genCacheKey(CacheConstants.TASK_EXTRACTOR_QUERY_TIME_MAP);
        RedissonClient redissonClient = RedissonClientProxy.getRedissonClient();
        RMap<Integer, Date> queryTimeMap = redissonClient.getMap(cacheKey);

        TaskQueryDto taskQueryDto = this.buildTaskQueryDto(taskGroup);
        if (exclusiveExecutableTask != null) {
            taskQueryDto.setExcludeTaskCodes(exclusiveExecutableTask.getTaskCodes());
        }
        //暂定设置20，分片节点每批次尽可能查询少量数据，这样可充分利用机器
        taskQueryDto.setTaskSizePerQuery(20);

        //获取集群节点最大开始时间
        Map<Integer, Date> timeMap = queryTimeMap.readAllMap();
        if (MapUtils.isNotEmpty(timeMap)) {
            Date startQueryTime = timeMap.values().stream().sorted(Comparator.reverseOrder()).findFirst().get();
            taskQueryDto.setStartTime(startQueryTime);
        }

        TaskPollResult taskPollResult = this.collectTask(taskGroup, taskQueryDto);
        if (taskPollResult == null || CollectionUtils.isEmpty(taskPollResult.getExecutableTasks())) {
            return taskPollResult;
        }

        int taskSize = taskPollResult.getExecutableTasks().size();
        TaskWrapper lastTaskWrapper = taskPollResult.getExecutableTasks().get(taskSize - 1);
        Date maximumExecuteTime = lastTaskWrapper.getEarliestExecuteTime();
        queryTimeMap.put(workNode.getRealWorkNode().getNodeNum(), maximumExecuteTime);
        queryTimeMap.expire(30, TimeUnit.MINUTES);

        return taskPollResult;
    }

    /**
     * 独享模式下，集群（任务组+任务）具有排他性，一个（任务组+任务）只能在一个节点上运行，需要正确配置；（目前未校验配置正确性）
     */
    private TaskPollResult collectExclusiveTask(TaskGroup taskGroup, WorkNode workNode) {
        WorkNodeExecutableTask workNodeExecutableTask = null;
        if (CollectionUtils.isNotEmpty(workNode.getMergedExecutableTasks())) {
            //只允许执行指定任务组任务
            workNodeExecutableTask = workNode.getMergedExecutableTasks().stream()
                    .filter(e -> e.getTaskGroup().equals(taskGroup.getGroup()))
                    .findFirst().orElse(null);
            if (workNodeExecutableTask == null) {
                return null;
            }
        }

        TaskQueryDto taskQueryDto = this.buildTaskQueryDto(taskGroup);
        if (workNodeExecutableTask != null) {
            //执行指定任务组特定任务，适用于DB队列；
            //如果想Redis队列达到同样效果，定义唯一任务组编码即可
            taskQueryDto.setTaskCodes(workNodeExecutableTask.getTaskCodes());
        }

        return this.collectTask(taskGroup, taskQueryDto);
    }

}
